/*******************************************************************************
 * Copyright (c) 2008 Sylvain Sicard. All rights reserved. This program and the accompanying materials are made available under the terms of the GNU Lesser
 * Public License v2.1 which accompanies this distribution, and is available at http://www.gnu.org/licenses/old-licenses/gpl-2.0.html
 * 
 * Contributors: Sylvain Sicard - initial API and implementation
 ******************************************************************************/
package fr.jade.event.filters;

import java.util.Vector;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import org.jgroups.Address;
import org.jgroups.Channel;
import org.jgroups.ChannelClosedException;
import org.jgroups.ChannelException;
import org.jgroups.ChannelNotConnectedException;
import org.jgroups.JChannelFactory;
import org.jgroups.Message;
import org.jgroups.Receiver;
import org.jgroups.View;
import org.objectweb.fractal.api.Component;
import org.objectweb.fractal.api.control.IllegalLifeCycleException;
import org.objectweb.fractal.api.control.LifeCycleController;

import fr.jade.event.Event;
import fr.jade.event.EventFactory;
import fr.jade.event.EventTag;
import fr.jade.reflex.util.Reflex;
import fr.jade.util.FractalNavigator;
import fr.jade.util.JadeException;

public class MasterSlaveEventFilterImpl extends BaseEventFilter implements LifeCycleController, Receiver {

  protected Channel channel;
  Vector<Address>   members    = new Vector<Address>();
  Address           local_addr = null;
  ScheduledExecutorService s = null;
  ScheduledFuture<?> handler = null;
  
  public MasterSlaveEventFilterImpl(){
    s = Executors.newScheduledThreadPool(1);
  }
  
  public void handleEvent(Event e) throws JadeException {

    if (isCoordinator()) {
      eh.handleEvent(e);
    } else {
      log("Discard event");
    }
  }

  protected void sendFailureEvent(Component failedCmp_M){
    assert failedCmp_M != null;
    Event e = EventFactory.newEvent(EventTag.FAILURE);
    e.getEventBody().put("failedCmp_M", failedCmp_M);
    try {
      //log("Send failure event");
      eh.handleEvent(e);
    } catch (JadeException e1) {
      log(e1);
    }
  }
  
  protected boolean isCoordinator() {
    Object first_mbr = null;

    synchronized (members) {
      first_mbr = !members.isEmpty() ? members.elementAt(0) : null;
      if (first_mbr == null)
        return true;
    }
    return getLocalAddress() != null && getLocalAddress().equals(first_mbr);
  }

  protected Address getLocalAddress() {
    if (local_addr != null)
      return local_addr;
    if (channel != null)
      local_addr = channel.getLocalAddress();
    return local_addr;
  }

  boolean started = false;

  @Override
  public String getFcState() {
    return started ? LifeCycleController.STARTED : LifeCycleController.STOPPED;
  }

  @Override
  public void startFc() throws IllegalLifeCycleException {
    if (!started) {
      if (channel == null) {
        JChannelFactory factoryStub = new JChannelFactory();
        try {
          factoryStub.setMultiplexerConfig("stacks.xml");
          channel = factoryStub.createChannel("udp");
          channel.setReceiver(this);
        } catch (Exception e) {
          throw new IllegalLifeCycleException(e.toString());
        }
      }
      try {
        channel.connect("jade");
      } catch (ChannelException e) {
        throw new IllegalLifeCycleException(e.toString());
      }
      handler = s.scheduleAtFixedRate(new T(), 5000, 5000, TimeUnit.MILLISECONDS);
      started = true;
    }
  }

  @Override
  public void stopFc() throws IllegalLifeCycleException {
    if (started) {
      handler.cancel(true);
      channel.disconnect();
      started = false;
    }
  }

  @Override
  public byte[] getState() {
    return null;
  }

  @Override
  public void receive(Message arg0) {
    
  }

  @Override
  public void setState(byte[] arg0) {

  }

  @Override
  public void block() {

  }

  @Override
  public void suspect(Address arg0) {
    try {
      Component node_E = FractalNavigator.getNode(self);
      Component node_M = Reflex.getDualComponent(node_E);
      sendFailureEvent(node_M);
    } catch (JadeException e) {
      e.printStackTrace();
    }
  }

  @Override
  public synchronized void viewAccepted(View new_view) {
    if (new_view == null)
      return;
    members = new_view.getMembers();
  }
  
  protected class T implements Runnable{

    public void run() {
      try {
        channel.send(new Message());
      } catch (ChannelNotConnectedException e) {
        e.printStackTrace();
      } catch (ChannelClosedException e) {
        e.printStackTrace();
      }
    }
    
  }

}
